package org.syx.edl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.syx.edl.nio.ByteLineReader;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteOrder;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author:whh
 * @date: 2025-09-05 23:13
 * <p></p>
 */
public class File2StorageBootStrap<I,O> implements AutoCloseable{

    private static final Logger log= LoggerFactory.getLogger(File2StorageBootStrap.class);
    private File file;
    private ItemProcessor<I,O> processor;

    private LineMapper<I> lineMapper;
    private ItemWriter<O> writer;
    private ItemValidator<I> validator;
    private List<ByteLineReader<I>> readers;
    private ExceptionHandler exceptionHandler;
    public static final int DEFAULT_BATCH = 1000;

    private int batch=DEFAULT_BATCH;


    public File2StorageBootStrap() {
        // 注册 JVM 关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(this::close));
    }

    public File2StorageBootStrap<I,O> file(File file) {
        this.file = file;
        return this;
    }

    public File2StorageBootStrap<I,O> writer(ItemWriter<O> writer) {
        this.writer = writer;
        return this;
    }
    public File2StorageBootStrap<I,O> processor(ItemProcessor<I,O> processor)  {
        this.processor = processor;
        return this;
    }
    public File2StorageBootStrap<I,O> lineMapper(LineMapper<I> lineMapper)  {
        this.lineMapper = lineMapper;
        return this;
    }
    public File2StorageBootStrap<I,O> validator(ItemValidator<I> validator) throws Exception {
        this.validator = validator;
        this.validator.open();
        return this;
    }

    public File2StorageBootStrap<I,O> exceptionHandler(ExceptionHandler exceptionHandler){
        this.exceptionHandler = exceptionHandler;
        return this;
    }


    public File2StorageBootStrap<I,O> chunks() throws IOException {
        this.readers= chunks(file).stream().map(buf-> new ByteLineReader<I>(buf,lineMapper)).collect(Collectors.toList());
        return this;
    }


    public void handle(){


        List<Exception> exceptionList = this.readers.stream().parallel().map(r -> {
            List<Exception> exp = new ArrayList<>();

            try {
                r.open();
                I item;
                while (r.hasNext()){
                    try {
                        item = r.next();
                        if (item != null) {
                            validator.validate(item);
                        }
                    } catch (Exception e) {
                        exp.add(e);
                    }
                }

            } catch (Exception e) {
                exp.add(e);
            }
            return exp;
        }).flatMap(List::stream).collect(Collectors.toList());


        if(!exceptionList.isEmpty()){
            exceptionHandler.handleException(exceptionList);
            return;
        }


        this.readers.stream().parallel().forEach(r->{

            try {
                r.flip();
                List<O> data=new ArrayList<>(batch);
                I item;
                while (r.hasNext()){
                    item = r.next();
                    if (item != null) {
                        O o = processor.process(item);
                        data.add(o);
                        if(data.size()>=batch){
                            writer.write(data);
                            data=new ArrayList<>(batch);
                        }
                    }
                }
                if(!data.isEmpty()){
                    writer.write(data);
                }
            } catch (Exception e) {
                log.info("数据处理异常",e);
                throw new RuntimeException(e);
            }
        });

    }
    @Override
    public void close()  {
        for (ItemReader<I> r : this.readers) {
            try {
                r.close();
            } catch (Exception e) {
                log.error("ItemReader关闭异常",e);
            }
        }
        if(writer!=null){
            try {
                writer.close();
            } catch (Exception e) {
                log.error("ItemWriter关闭异常",e);
            }
        }
        if(validator!=null){
            try {
                validator.close();
            } catch (Exception e) {
                log.error("ItemValidator关闭异常",e);
            }
        }
    }



    public static final long DEFAULT_CHUNK = 1024 * 1024 * 100;//100MB  1024 * 1024 * 100

    private static long chunk = DEFAULT_CHUNK;


    /**
     * 对文件进行分块拆分以行分隔符为结尾
     * @param file
     * @return
     * @throws IOException
     */
   private List<MappedByteBuffer> chunks(File file) throws IOException {
        try(RandomAccessFile r=new RandomAccessFile(file,"r");) {
            long length = r.length();
            List<MappedByteBuffer> chunks=new ArrayList<>((int) (length/chunk)+1);
            long start=0L;
            while (start<length){
                long pos=start+chunk;
                if(pos<length){
                    r.seek(pos);
                    do{
                        pos+=1;
                    }while (r.read()!='\n' && pos<length);
                }else {
                    pos=length;
                }
                MappedByteBuffer buffer = r.getChannel().map(FileChannel.MapMode.READ_ONLY, start, pos - start);
                buffer.order(ByteOrder.nativeOrder());
                chunks.add(buffer);
                start=pos;
            }

            return chunks;
        }catch (IOException e){
            log.error("文件拆分异常",e);
            throw e;
        }

    }

}
